#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time    : 2025/8/19
# @USER    : Shengji He
# @File    : network.py
# @Software: PyCharm
# @Version  : Python-
# @TASK:
import math
import torch
import torch.nn as nn
import torch.nn.functional as F

__all__ = ['VQGAN', 'Discriminator', 'Transformer']


class Encoder(nn.Module):
    def __init__(self):
        super().__init__()
        # (B, 1, 28, 28)>> (B, 32, 14, 14)>> (B, 16, 7, 7)
        self.net = nn.Sequential(
            nn.Conv2d(in_channels=1, out_channels=32, kernel_size=4, stride=2, padding=1),
            nn.ReLU(),
            nn.Dropout(0.3),
            nn.Conv2d(32, 16, 4, 2, 1),
            nn.ReLU()
        )
        self.set_para()

    def set_para(self):
        nn.init.xavier_normal_(self.net[0].weight, nn.init.calculate_gain('relu'))
        nn.init.xavier_normal_(self.net[3].weight, nn.init.calculate_gain('relu'))

    def forward(self, x):
        return self.net(x)


class Decoder(nn.Module):
    def __init__(self):
        super().__init__()
        self.net = nn.Sequential(
            nn.ConvTranspose2d(16, 32, 4, 2, 1),
            nn.ReLU(),
            nn.ConvTranspose2d(32, 1, 4, 2, 1),
            nn.Sigmoid()
        )
        self.set_para()

    def set_para(self):
        nn.init.xavier_normal_(self.net[0].weight, nn.init.calculate_gain('relu'))
        nn.init.xavier_normal_(self.net[2].weight, nn.init.calculate_gain('sigmoid'))

    def forward(self, x):
        return self.net(x)


class Codebook(nn.Module):
    def __init__(self, num_codebook_vectors=20, latent_dim=16):
        super(Codebook, self).__init__()
        self.num_codebook_vectors, self.latent_dim = num_codebook_vectors, latent_dim
        self.beta = 0.25
        self.embedding = nn.Embedding(self.num_codebook_vectors, self.latent_dim)
        self.set_para()

    def set_para(self):
        nn.init.xavier_normal_(self.embedding.weight)

    def forward(self, z):
        # (B, 16, 7, 7)>> (B, 7, 7, 16)
        z = z.permute(0, 2, 3, 1).contiguous()
        # (B* 7* 7, 16)
        z_flattened = z.view(-1, self.latent_dim)
        # (B* 7* 7, 1)+ (20, )- (B* 7* 7, 20)= (49B, 20)
        d = (torch.sum(z_flattened ** 2, dim=1, keepdim=True) +
             torch.sum(self.embedding.weight ** 2, dim=1) -
             2 * torch.matmul(z_flattened, self.embedding.weight.T)
             )
        # (49B,)
        min_encoding_indices = torch.argmin(d, dim=1)
        # (49B, 16)>> (B, 7, 7, 16)
        z_q = self.embedding(min_encoding_indices).view(z.shape)
        loss = torch.mean((z_q.detach() - z) ** 2) + self.beta * torch.mean((z_q - z.detach()) ** 2)
        z_q = z + (z_q - z).detach()
        # (B, 16, 7, 7)
        z_q = z_q.permute(0, 3, 1, 2)
        # (B, 16, 7, 7), (49B,).
        return z_q, min_encoding_indices, loss


class VQGAN(nn.Module):
    def __init__(self, num_codebook_vectors=20, latent_dim=16):
        super(VQGAN, self).__init__()
        self.encoder = Encoder()
        self.decoder = Decoder()
        self.codebook = Codebook(num_codebook_vectors, latent_dim)
        self.quant_conv = nn.Conv2d(16, 16, 1)
        self.post_quant_conv = nn.Conv2d(16, 16, 1)
        self.set_para()

    def set_para(self):
        nn.init.xavier_normal_(self.quant_conv.weight)
        nn.init.xavier_normal_(self.post_quant_conv.weight)

    def mse_loss(self, label, pred):
        return ((label - pred) ** 2).mean()

    def forward(self, x):
        encoded_images = self.encoder(x)
        quant_conv_encoded_images = self.quant_conv(encoded_images)
        codebook_mapping, codebook_indices, q_loss = self.codebook(quant_conv_encoded_images)
        post_quant_conv_mapping = self.post_quant_conv(codebook_mapping)
        decoded_images = self.decoder(post_quant_conv_mapping)
        return decoded_images, codebook_indices, q_loss

    def encode(self, x):
        # (B, 16, 7, 7)
        encoded_images = self.encoder(x)
        # (B, 16, 1, 1)
        quant_conv_encoded_images = self.quant_conv(encoded_images)
        #
        codebook_mapping, codebook_indices, q_loss = self.codebook(quant_conv_encoded_images)
        return codebook_mapping, codebook_indices, q_loss

    def decode(self, z):
        post_quant_conv_mapping = self.post_quant_conv(z)
        decoded_images = self.decoder(post_quant_conv_mapping)
        return decoded_images

    def calculate_lambda(self, recon_loss, gan_loss):
        last_layer = self.decoder.net[-2]
        last_layer_weight = last_layer.weight
        recon_loss_grads = torch.autograd.grad(recon_loss, last_layer_weight, retain_graph=True)[0]
        gan_loss_grads = torch.autograd.grad(gan_loss, last_layer_weight, retain_graph=True)[0]
        λ = torch.norm(recon_loss_grads) / (torch.norm(gan_loss_grads) + 1e-10)
        return 0.8 * torch.clamp(λ, 0, 1e4).detach()

    @staticmethod
    def adopt_weight(disc_factor, i, threshold, value=0.):
        if i < threshold:
            disc_factor = value
        return disc_factor


class Discriminator(nn.Module):
    def __init__(self):
        super().__init__()
        self.cv = nn.Sequential(
            nn.Conv2d(in_channels=1, out_channels=32, kernel_size=4, stride=2, padding=1),
            nn.ReLU(),
            nn.Dropout(0.3),
            nn.Conv2d(32, 16, 4, 2, 1),
            nn.ReLU(),
            nn.Dropout(0.3),
            nn.Conv2d(16, 1, 1),
            nn.Sigmoid()
        )
        self.set_para()

    def set_para(self):
        nn.init.xavier_normal_(self.cv[0].weight, nn.init.calculate_gain('relu'))
        nn.init.xavier_normal_(self.cv[3].weight, nn.init.calculate_gain('relu'))
        nn.init.xavier_normal_(self.cv[6].weight, nn.init.calculate_gain('sigmoid'))

    def forward(self, x):
        return self.cv(x)


"""
taken from: https://github.com/karpathy/minGPT/
GPT model:
- the initial stem consists of a combination of token encoding and a positional encoding
- the meat of it is a uniform sequence of Transformer blocks
    - each Transformer is a sequential combination of a 1-hidden-layer MLP block and a self-attention block
    - all blocks feed into a central residual pathway similar to resnets
- the final decoder is a linear projection into a vanilla Softmax classifier
"""


class GPTConfig:
    """ base GPT config, params common to all GPT versions """
    embd_pdrop = 0.1
    resid_pdrop = 0.1
    attn_pdrop = 0.1

    # 20, 16
    def __init__(self, vocab_size, block_size, **kwargs):
        self.vocab_size = vocab_size
        self.block_size = block_size
        for k, v in kwargs.items():
            # 设置属性值
            setattr(self, k, v)


class CausalSelfAttention(nn.Module):
    """
    A vanilla multi-head masked self-attention layer with a projection at the end.
    It is possible to use torch.nn.MultiheadAttention here but I am including an
    explicit implementation here to show that there is nothing too scary here.
    """

    def __init__(self, config):
        super().__init__()
        assert config.n_embd % config.n_head == 0
        # key, query, value projections for all heads
        self.key = nn.Linear(config.n_embd, config.n_embd)
        self.query = nn.Linear(config.n_embd, config.n_embd)
        self.value = nn.Linear(config.n_embd, config.n_embd)
        # regularization
        self.attn_drop = nn.Dropout(config.attn_pdrop)
        self.resid_drop = nn.Dropout(config.resid_pdrop)
        # output projection
        self.proj = nn.Linear(config.n_embd, config.n_embd)
        # causal mask to ensure that attention is only applied to the left in the input sequence
        mask = torch.tril(torch.ones(config.block_size,
                                     config.block_size))
        if hasattr(config, "n_unmasked"):
            mask[:config.n_unmasked, :config.n_unmasked] = 1
        self.register_buffer("mask", mask.view(1, 1, config.block_size, config.block_size))
        self.n_head = config.n_head

    def forward(self, x, layer_past=None):
        B, T, C = x.size()

        # calculate query, key, values for all heads in batch and move head forward to be the batch dim
        k = self.key(x).view(B, T, self.n_head, C // self.n_head).transpose(1, 2)  # (B, nh, T, hs)
        q = self.query(x).view(B, T, self.n_head, C // self.n_head).transpose(1, 2)  # (B, nh, T, hs)
        v = self.value(x).view(B, T, self.n_head, C // self.n_head).transpose(1, 2)  # (B, nh, T, hs)

        present = torch.stack((k, v))
        if layer_past is not None:
            past_key, past_value = layer_past
            k = torch.cat((past_key, k), dim=-2)
            v = torch.cat((past_value, v), dim=-2)

        # causal self-attention; Self-attend: (B, nh, T, hs) x (B, nh, hs, T) -> (B, nh, T, T)
        att = (q @ k.transpose(-2, -1)) * (1.0 / math.sqrt(k.size(-1)))
        if layer_past is None:
            att = att.masked_fill(self.mask[:, :, :T, :T] == 0, float('-inf'))

        att = F.softmax(att, dim=-1)
        att = self.attn_drop(att)
        y = att @ v  # (B, nh, T, T) x (B, nh, T, hs) -> (B, nh, T, hs)
        y = y.transpose(1, 2).contiguous().view(B, T, C)  # re-assemble all head outputs side by side

        # output projection
        y = self.resid_drop(self.proj(y))
        return y, present  # TODO: check that this does not break anything


class Block(nn.Module):
    """ an unassuming Transformer block """

    def __init__(self, config):
        super().__init__()
        self.ln1 = nn.LayerNorm(config.n_embd)
        self.ln2 = nn.LayerNorm(config.n_embd)
        self.attn = CausalSelfAttention(config)
        self.mlp = nn.Sequential(
            nn.Linear(config.n_embd, 4 * config.n_embd),
            nn.GELU(),  # nice
            nn.Linear(4 * config.n_embd, config.n_embd),
            nn.Dropout(config.resid_pdrop),
        )

    def forward(self, x, layer_past=None, return_present=False):
        # TODO: check that training still works
        if return_present:
            assert not self.training
        # layer past: tuple of length two with B, nh, T, hs
        attn, present = self.attn(self.ln1(x), layer_past=layer_past)

        x = x + attn
        x = x + self.mlp(self.ln2(x))
        if layer_past is not None or return_present:
            return x, present
        return x


class GPT(nn.Module):
    """  the full GPT language model, with a context size of block_size """

    # 20,
    def __init__(self, vocab_size, block_size, n_layer=12, n_head=8, n_embd=256,
                 embd_pdrop=0., resid_pdrop=0., attn_pdrop=0., n_unmasked=0):
        super().__init__()
        config = GPTConfig(vocab_size=vocab_size, block_size=block_size,
                           embd_pdrop=embd_pdrop, resid_pdrop=resid_pdrop, attn_pdrop=attn_pdrop,
                           n_layer=n_layer, n_head=n_head, n_embd=n_embd,
                           n_unmasked=n_unmasked)
        # input embedding stem
        self.tok_emb = nn.Embedding(config.vocab_size, config.n_embd)
        self.pos_emb = nn.Parameter(torch.zeros(1, config.block_size, config.n_embd))
        self.drop = nn.Dropout(config.embd_pdrop)
        # transformer
        self.blocks = nn.Sequential(*[Block(config) for _ in range(config.n_layer)])
        # decoder head
        self.ln_f = nn.LayerNorm(config.n_embd)
        self.head = nn.Linear(config.n_embd, config.vocab_size, bias=False)

        self.block_size = config.block_size
        self.apply(self._init_weights)
        self.config = config

    def get_block_size(self):
        return self.block_size

    def _init_weights(self, module):
        if isinstance(module, (nn.Linear, nn.Embedding)):
            module.weight.data.normal_(mean=0.0, std=0.02)
            if isinstance(module, nn.Linear) and module.bias is not None:
                module.bias.data.zero_()
        elif isinstance(module, nn.LayerNorm):
            module.bias.data.zero_()
            module.weight.data.fill_(1.0)

    def forward(self, idx, embeddings=None):
        # idx, (B, 49)>> (B, 49, 16)
        token_embeddings = self.tok_emb(idx)  # each index maps to a (learnable) vector
        if embeddings is not None:  # prepend explicit embeddings
            token_embeddings = torch.cat((embeddings, token_embeddings), dim=1)
        # t= 49
        t = token_embeddings.shape[1]
        # 49<= 49
        assert t <= self.block_size, "Cannot forward, model block size is exhausted."
        # (B, 49, 16)
        position_embeddings = self.pos_emb[:, :t, :]  # each position maps to a (learnable) vector
        x = self.drop(token_embeddings + position_embeddings)
        x = self.blocks(x)
        x = self.ln_f(x)
        # (B, 49, 20)
        logits = self.head(x)
        return logits, None


class Transformer(nn.Module):
    def __init__(self, vqgan):
        super(Transformer, self).__init__()
        self.sos_token = 0
        self.vqgan = vqgan
        # codebook vector num, block_size , layer num, head, embed num
        self.transformer = GPT(30, 49, 2, 4, 16)
        self.pkeep = 0.5

    def encode_to_z(self, x):
        # (B, 16, 7, 7), (49B,),
        quant_z, indices, _ = self.vqgan.encode(x)
        # (B, 49)
        indices = indices.view(quant_z.shape[0], -1)
        return quant_z, indices

    def z_to_image(self, indices):
        # (B, 49, 16)>> (B, 7, 7, 16)
        ix_to_vec = self.vqgan.codebook.embedding(indices).reshape(indices.shape[0], 7, 7, 16)
        # (B, 16, 7, 7)
        ix_to_vec = ix_to_vec.permute(0, 3, 1, 2)
        # (B, 1, 28, 28)
        image = self.vqgan.decode(ix_to_vec)
        # (28, 28, 1)
        return (image[0]).permute(1, 2, 0)

    def forward(self, x):
        # x, (B, 1, 28, 28); indices, (B, 49)
        _, indices = self.encode_to_z(x)
        # (B, 1)
        sos_tokens = torch.ones(x.shape[0], 1) * self.sos_token
        sos_tokens = sos_tokens.long().to(x.device)
        # mask, (B, 49)
        mask = torch.bernoulli(self.pkeep * torch.ones(indices.shape, device=indices.device))
        mask = mask.round().to(dtype=torch.int64)
        # value: 0- 19; size: (B, 49).
        random_indices = torch.randint_like(indices, 20)
        new_indices = mask * indices + (1 - mask) * random_indices
        # new_indices, (B, 50)
        new_indices = torch.cat((sos_tokens, new_indices), dim=1)
        # (B, 49)
        target = indices
        # (B, 49, 20),
        logits, _ = self.transformer(new_indices[:, :-1])
        # (B, 49, 20); (B, 49)
        return logits, target

    def top_k_logits(self, logits, k):
        # logits, (B, 49, 20); k, 2.
        # v, (B, 49, 2); ix, (B, 49, 2).
        v, ix = torch.topk(logits, k)
        out = logits.clone()
        # mask on out, < top k.
        out[out < v[:, [-1]]] = -float('inf')
        return out

    @torch.no_grad()
    def sample(self, x, c, steps, temperature=1.0, topk=10):
        # x, (B, 24); c, (B, 1); steps, 25.
        self.transformer.eval()
        # x, (B, 25).
        x = torch.cat((c, x), dim=1)
        # for each step
        for k in range(steps):
            # x, (B, 25); logits, (B, 25, 20).
            logits, _ = self.transformer(x)
            # (B, 1, 20)
            logits = logits[:, -1, :] / temperature
            if topk is not None:
                logits = self.top_k_logits(logits, topk)
            # (B, 1, 20)
            probs = F.softmax(logits, dim=-1)
            # (B, 1)
            ix = torch.multinomial(probs, num_samples=1)
            # (B, 26)
            x = torch.cat((x, ix), dim=1)
        # (B, 49)
        x = x[:, c.shape[1]:]
        self.transformer.train()
        return x

    @torch.no_grad()
    def log_images(self, x):
        log = dict()
        # (B, 49)
        _, indices = self.encode_to_z(x)
        # (B, 1)
        sos_tokens = torch.ones(x.shape[0], 1) * self.sos_token
        sos_tokens = sos_tokens.long().to(x.device)
        # (B, 24)
        start_indices = indices[:, :indices.shape[1] // 2]
        # (B, 24); (B, 1); 49- 24= 25>> (B, 49)
        sample_indices = self.sample(start_indices, sos_tokens, steps=indices.shape[1] - start_indices.shape[1])
        # (B, 49)>> (B, 1, 28, 28)
        half_sample = self.z_to_image(sample_indices)
        start_indices = indices[:, :0]
        sample_indices = self.sample(start_indices, sos_tokens, steps=indices.shape[1])
        full_sample = self.z_to_image(sample_indices)
        x_rec = self.z_to_image(indices)
        log['input'] = x
        log['rec'] = x_rec
        log['half_sample'] = half_sample
        log['full_sample'] = full_sample
        return log, torch.stack((x[0].permute(1, 2, 0), x_rec, half_sample, full_sample))
